package io.kiki.sba.registry.server.clientInterface.resource;

import com.codahale.metrics.Gauge;
import com.codahale.metrics.MetricRegistry;
import io.kiki.sba.registry.api.exchange.NodeExchanger;
import io.kiki.sba.registry.api.exchange.message.SimpleRequest;
import io.kiki.sba.registry.common.model.ClientToClientInterfaceServerNodeId;
import io.kiki.sba.registry.common.model.GenericResponse;
import io.kiki.sba.registry.common.model.Response;
import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.appmeta.InterfaceMapping;
import io.kiki.sba.registry.common.model.sessionserver.PubSubDataInfoIdRequest;
import io.kiki.sba.registry.common.model.sessionserver.PubSubDataInfoIdResp;
import io.kiki.sba.registry.common.model.store.*;
import io.kiki.sba.registry.metrics.ReporterUtils;
import io.kiki.sba.registry.net.NetUtil;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.clientInterface.providedata.FetchStopPushService;
import io.kiki.sba.registry.server.clientInterface.store.FetchPubSubDataInfoIdService;
import io.kiki.sba.registry.server.clientInterface.store.Publishers;
import io.kiki.sba.registry.server.clientInterface.store.Subscribers;
import io.kiki.sba.registry.server.clientInterface.store.Watchers;
import io.kiki.sba.registry.server.shared.meta.MetaServerService;
import io.kiki.sba.registry.store.api.repository.AppRevisionRepository;
import io.kiki.sba.registry.store.api.repository.InterfaceAppsRepository;
import io.kiki.sba.registry.task.MetricsableThreadPoolExecutor;
import io.kiki.sba.registry.util.OsUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.ws.rs.*;
import javax.ws.rs.core.MediaType;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ThreadPoolExecutor;

import static io.kiki.sba.registry.common.model.constants.ValueConstants.CONNECT_ID_SPLIT;


@Path("digest")
public class SessionDigestResource {

    private static final Logger logger = LoggerFactory.getLogger(ClientManagerResource.class);
    private static final String LOCAL_ADDRESS = NetUtil.getLocalAddress().getHostAddress();
    private static final String SUB = "SUB";
    private static final String PUB = "PUB";
    private static final String WAT = "WAT";
    private static final String SESSION = "SESSION";
    private static final String DATA = "DATA";
    private static final String META = "META";
    private final ThreadPoolExecutor pubSubQueryZoneExecutor = MetricsableThreadPoolExecutor.newExecutor("PubSubQueryZoneExecutor", OsUtils.getCpuCount(), 100, new ThreadPoolExecutor.CallerRunsPolicy());
    /**
     * store subscribers
     */
    @Autowired
    private Subscribers sessionSubscribers;
    /**
     * store watchers
     */
    @Autowired
    private Watchers sessionWatchers;
    /**
     * store publishers
     */
    @Autowired
    private Publishers sessionPublishers;
    @Autowired
    private MetaServerService metaNodeService;
    @Autowired
    private Config config;
    @Resource
    private FetchStopPushService fetchStopPushService;
    @Autowired
    private FetchPubSubDataInfoIdService fetchPubSubDataInfoIdService;
    @Autowired
    private NodeExchanger sessionConsoleExchanger;
    @Autowired
    private AppRevisionRepository appRevisionRepository;
    @Autowired
    private InterfaceAppsRepository interfaceAppsRepository;
    @Autowired
    private MetaServerService mataNodeService;

    @PostConstruct
    public void init() {
        MetricRegistry metrics = new MetricRegistry();
        metrics.register("pushSwitch", (Gauge<Map>) () -> getPushSwitch());
        ReporterUtils.startSlf4jReporter(60, metrics);
    }

    @GET
    @Path("/metadata/allRevisionIds")
    @Produces(MediaType.APPLICATION_JSON)
    public Set<String> allRevisionIds() {
        return appRevisionRepository.allRevisionIds();
    }

    @GET
    @Path("/metadata/allServiceMapping")
    @Produces(MediaType.APPLICATION_JSON)
    public Map<String, Map<String, InterfaceMapping>> allServiceMapping() {
        return interfaceAppsRepository.allServiceMapping();
    }

    @GET
    @Path("{type}/data/query")
    @Produces(MediaType.APPLICATION_JSON)
    public Map<String, Collection<? extends StoreData>> getSessionDataByDataInfoId(@QueryParam("dataInfoId") String dataInfoId, @PathParam("type") String type) {
        Map<String, Collection<? extends StoreData>> serverList = new HashMap<>();
        if (dataInfoId != null) {
            Collection<Publisher> publishers = sessionPublishers.getDataList(dataInfoId);
            Collection<Subscriber> subscribers = sessionSubscribers.getDataList(dataInfoId);
            Collection<Watcher> watchers = sessionWatchers.getDataList(dataInfoId);
            fillServerList(type, serverList, publishers, subscribers, watchers);
        }

        return serverList;
    }

    @POST
    @Path("{type}/connect/query")
    @Produces(MediaType.APPLICATION_JSON)
    public Map<String, Collection<? extends StoreData>> getSessionDataByConnectId(List<String> queryConnectIds, final @PathParam("type") String type) {
        List<ClientToClientInterfaceServerNodeId> clientToClientInterfaceServerNodeIds = new ArrayList<>(queryConnectIds.size());
        for (String queryConnectId : queryConnectIds) {
            String connectId = queryConnectId;
            if (!queryConnectId.contains(CONNECT_ID_SPLIT)) {
                connectId = connectId + CONNECT_ID_SPLIT + LOCAL_ADDRESS + ":" + config.getServerPort();
            }
            clientToClientInterfaceServerNodeIds.add(ClientToClientInterfaceServerNodeId.parse(connectId));
        }

        Map<String, Collection<? extends StoreData>> serverList = new HashMap<>();

        clientToClientInterfaceServerNodeIds.forEach(connectId -> {
            Map pubMap = sessionPublishers.queryByConnectId(connectId);
            Map subMap = sessionSubscribers.queryByConnectId(connectId);
            Map watcherMap = sessionWatchers.queryByConnectId(connectId);

            Collection<Publisher> publishers = pubMap != null && !pubMap.isEmpty() ? pubMap.values() : new ArrayList<>();
            Collection<Subscriber> subscribers = subMap != null && !subMap.isEmpty() ? subMap.values() : new ArrayList<>();
            Collection<Watcher> watchers = watcherMap != null && !watcherMap.isEmpty() ? watcherMap.values() : new ArrayList<>();
            fillServerList(type, serverList, publishers, subscribers, watchers);
        });

        return serverList;
    }

    @GET
    @Path("/data/queryDetail")
    @Produces(MediaType.APPLICATION_JSON)
    public GenericResponse<PubSubDataInfoIdResp> queryDetail(@QueryParam("ips") String ips) {
        HashSet<String> ipSet = new HashSet<>(Arrays.asList(ips.split(",")));
        if (CollectionUtils.isEmpty(ipSet)) {
            return new GenericResponse().fillFailed("param ips is empty");
        }
        PubSubDataInfoIdResp localResp = fetchPubSubDataInfoIdService.queryByIps(ipSet);
        List<Url> servers = Sdks.getOtherConsoleServers(null, config, metaNodeService);

        List<PubSubDataInfoIdResp> resps = new ArrayList<>(servers.size() + 1);
        resps.add(localResp);
        if (servers.size() > 0) {
            Map<Url, Response> map = Sdks.concurrentSdkSend(pubSubQueryZoneExecutor, servers, (Url url) -> {
                final PubSubDataInfoIdRequest req = new PubSubDataInfoIdRequest(ipSet);
                return (Response) sessionConsoleExchanger.request(new SimpleRequest(req, url)).getResult();
            }, 5000);

            for (Entry<Url, Response> entry : map.entrySet()) {
                if (entry.getValue() instanceof GenericResponse) {
                    GenericResponse response = (GenericResponse) entry.getValue();
                    if (response.isSuccess()) {
                        resps.add((PubSubDataInfoIdResp) response.getData());
                        continue;
                    }
                }
                logger.error("url={} query pub and sub dataInfoIds fail, msg:{}.", entry.getKey().getIp(), entry.getValue());
            }
        }
        return new GenericResponse<PubSubDataInfoIdResp>().fillSucceed(merge(resps));
    }

    protected PubSubDataInfoIdResp merge(List<PubSubDataInfoIdResp> resps) {
        if (resps.size() == 1) {
            return resps.get(0);
        }

        Map<String, Set<String>> pubDataInfoIds = new HashMap<>();
        Map<String, Set<String>> subDataInfoIds = new HashMap<>();
        for (PubSubDataInfoIdResp resp : resps) {
            if (!CollectionUtils.isEmpty(resp.getPubDataInfoIds())) {
                for (Entry<String, Set<String>> pubEntry : resp.getPubDataInfoIds().entrySet()) {
                    Set<String> set = pubDataInfoIds.computeIfAbsent(pubEntry.getKey(), k -> new HashSet<>());
                    set.addAll(pubEntry.getValue());
                }
            }

            if (!CollectionUtils.isEmpty(resp.getSubDataInfoIds())) {
                for (Entry<String, Set<String>> subEntry : resp.getSubDataInfoIds().entrySet()) {
                    Set<String> set = subDataInfoIds.computeIfAbsent(subEntry.getKey(), k -> new HashSet<>());
                    set.addAll(subEntry.getValue());
                }
            }
        }
        return new PubSubDataInfoIdResp(pubDataInfoIds, subDataInfoIds);
    }

    @GET
    @Path("/data/count")
    @Produces(MediaType.APPLICATION_JSON)
    public String getSessionDataCount() {
        Tuple<Long, Long> countSub = sessionSubscribers.count();
        Tuple<Long, Long> countPub = sessionPublishers.count();
        Tuple<Long, Long> countSubW = sessionWatchers.count();

        return String.format("Subscriber count: %s, Publisher count: %s, Watcher count: %s", countSub.o2, countPub.o2, countSubW.o2);
    }

    @GET
    @Path("pushSwitch")
    @Produces(MediaType.APPLICATION_JSON)
    public Map<String, Object> getPushSwitch() {
        Map<String, Object> resultMap = new HashMap<>(1);
        resultMap.put("pushSwitch", !fetchStopPushService.isStopPushSwitch() ? "open" : "closed");
        return resultMap;
    }

    @GET
    @Path("getDataInfoIdList")
    @Produces(MediaType.APPLICATION_JSON)
    public Collection<String> getDataInfoIdList() {
        Collection<String> ret = new HashSet<>();
        ret.addAll(sessionSubscribers.getDataInfoIds());
        ret.addAll(sessionPublishers.getDataInfoIds());
        return ret;
    }

    @GET
    @Path("{type}/getDataInfoIdListGroupByCount")
    @Produces(MediaType.APPLICATION_JSON)
    public List<Map.Entry<String, Integer>> getDataInfoIdListGroupByCount(final @PathParam("type") String type) {
        List<Map.Entry<String, Integer>> list = new ArrayList<>();
        String inputType = type.toUpperCase();
        switch (inputType) {
        case PUB:
            sessionPublishers.getDatas().forEach((key, value) -> list.add(new AbstractMap.SimpleEntry<>(key, value.size())));
            break;
        case WAT:
            sessionWatchers.getDatas().forEach((key, value) -> list.add(new AbstractMap.SimpleEntry<>(key, value.size())));
            break;
        default:
            sessionSubscribers.getDatas().forEach((key, value) -> list.add(new AbstractMap.SimpleEntry<>(key, value.size())));
            break;
        }
        list.sort(Collections.reverseOrder(Map.Entry.comparingByValue()));
        return list;
    }

    @GET
    @Path("checkSumDataInfoIdList")
    @Produces(MediaType.APPLICATION_JSON)
    public int checkSumDataInfoIdList() {
        return getDataInfoIdList().hashCode();
    }

    private void fillServerList(String type, Map<String, Collection<? extends StoreData>> serverList, Collection<Publisher> publishers, Collection<Subscriber> subscribers, Collection<Watcher> watchers) {
        if (type != null && !type.isEmpty()) {
            String inputType = type.toUpperCase();

            switch (inputType) {
            case PUB:
                if (!CollectionUtils.isEmpty(publishers)) {
                    serverList.put(PUB, publishers);
                }
                break;
            case SUB:
                if (!CollectionUtils.isEmpty(subscribers)) {
                    serverList.put(SUB, subscribers);
                }
                break;
            case WAT:
                if (!CollectionUtils.isEmpty(watchers)) {
                    serverList.put(WAT, watchers);
                }
                break;
            default:
                if (!CollectionUtils.isEmpty(publishers)) {
                    serverList.put(PUB, publishers);
                }
                if (!CollectionUtils.isEmpty(subscribers)) {
                    serverList.put(SUB, subscribers);
                }
                if (!CollectionUtils.isEmpty(watchers)) {
                    serverList.put(WAT, watchers);
                }
                break;
            }

        } else {
            if (publishers != null) {
                serverList.put(PUB, publishers);
            }
            if (subscribers != null) {
                serverList.put(SUB, subscribers);
            }
            if (watchers != null) {
                serverList.put(WAT, watchers);
            }
        }
    }

    @GET
    @Path("{type}/serverList/query")
    @Produces(MediaType.APPLICATION_JSON)
    public List<String> getServerListAll(@PathParam("type") String type) {
        List<String> serverList = new ArrayList<>();
        if (type != null && !type.isEmpty()) {
            String inputType = type.toUpperCase();

            switch (inputType) {
            case SESSION:
                serverList = getSessionServerList();
                break;
            case DATA:
                serverList = getDataServerList();
                break;
            case META:
                serverList = getMetaServerLeader();
                break;
            default:
                serverList = new ArrayList<>();
                break;
            }
        }
        return serverList;
    }

    public List<String> getSessionServerList() {
        return mataNodeService.getSessionServerList(config.getSessionServerRegion());
    }

    public List<String> getDataServerList() {
        return new ArrayList<>(mataNodeService.getDataServerList());
    }

    public List<String> getMetaServerLeader() {
        return new ArrayList<>(Arrays.asList(mataNodeService.getMetaServerLeader()));
    }
}
